package com.its.common.kafka.config;

import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.util.StringUtils;

import java.util.Properties;

@Data
@Slf4j
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class KafkaConsumerProperties {

    private String bootstrapServers;

    private String topic;

    private String groupId;

    private String securityProtocol; // "SASL_PLAINTEXT"

    private String mechanism; // "SCRAM-SHA-256"

    private String username;

    private String password;

    private Integer workerNum;

    private String workerClass;

    public Class getWorkerClaz() {
        Class claz = null;
        try {
            claz = Class.forName(workerClass);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("Kafka消费者类文件找不到!");
        }
        return claz;
    }

    public Properties buildProperties(){
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "60000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        if(StringUtils.hasText(this.getSecurityProtocol()) && this.getSecurityProtocol().length() > 5) {
//            properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
//            properties.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");

            props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, this.getSecurityProtocol());
            props.put(SaslConfigs.SASL_MECHANISM, this.getMechanism());

            if(StringUtils.hasText(this.getUsername())){
                props.put(SaslConfigs.SASL_JAAS_CONFIG,
                        "org.apache.kafka.common.security.scram.ScramLoginModule required "
                                + "username=\"" + this.getUsername() + "\" "
                                + "password=\"" + this.getPassword() +"\";");
            }else{
                throw new RuntimeException("当前Kafka的安全协议设置为：SASL_PLAINTEXT，但没有设置用户信息");
            }
        }

        return props;
    }
}
